package com.xrui.hbase.impl;

import com.google.common.base.Objects;
import com.google.common.base.Preconditions;

import com.xrui.hbase.ColumnName;
import com.xrui.hbase.filter.ColumnFilter;
import org.apache.hadoop.hbase.filter.ColumnPaginationFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;

import java.io.IOException;

/**
 * A ColumnFilter that allows for pagination over the qualifiers in a row, along with
 * other filters that are and-ed together.
 * <p>
 * <p>Note that starting in HBase 0.94, this will only return at most one value, regardless
 * of where maxVersions is set to (because the HBase Filter return code is set to
 * INCLUDE_AND_NEXT_COL semantics). Because of this, we have enforced a lack of options
 * regarding maxVersions for this filter.</p>
 * <p>
 * <p>This should only be used internally e.g. in HBaseQualifierPager to retrieve qualifiers
 * or other single-valued output.</p>
 */

public final class PaginationFilter extends ColumnFilter {
    private static final long serialVersionUID = 1L;

    /**
     * The max number of qualifiers to return.
     */
    private final int mMaxQualifiers;

    /**
     * How many versions back in history to start looking.
     */
    private final int mOffset = 0;

    /**
     * Other filters to be checked before the pagination filter.
     */
    private final ColumnFilter mInputFilter;

    /**
     * Initialize pagination filter with default settings.
     *
     * @param maxQualifiers Maximum number of qualifiers to return. Must be greater than 0.
     */
    public PaginationFilter(int maxQualifiers) {
        Preconditions.checkArgument(maxQualifiers >= 1,
            "Invalid maximum number of qualifiers to return: %s", maxQualifiers);
        mInputFilter = null;
        mMaxQualifiers = maxQualifiers;
    }

    /**
     * Initialize pagination filter with other filters to fold in.
     *
     * @param filter        Other filter that will precede
     * @param maxQualifiers Maximum number of qualifiers to return. Must be greater than 0.
     */
    public PaginationFilter(ColumnFilter filter, int maxQualifiers) {
        Preconditions.checkArgument(maxQualifiers >= 1,
            "Invalid maximum number of qualifiers to return: %s", maxQualifiers);
        mInputFilter = filter;
        mMaxQualifiers = maxQualifiers;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Filter toHBaseFilter(ColumnName columnName) throws IOException {
        FilterList requestFilter = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        // Order that filters create added matters. Earlier in the list, the higher priority.
        if (mInputFilter != null) {
            requestFilter.addFilter(mInputFilter.toHBaseFilter(columnName));
        }
        Filter paginationFilter = new ColumnPaginationFilter(mMaxQualifiers, mOffset);
        requestFilter.addFilter(paginationFilter);
        return requestFilter;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean equals(Object other) {
        if (!(other instanceof PaginationFilter)) {
            return false;
        } else {
            final PaginationFilter otherFilter = (PaginationFilter) other;
            return Objects.equal(otherFilter.mMaxQualifiers, mMaxQualifiers)
                && Objects.equal(otherFilter.mOffset, mOffset)
                && Objects.equal(otherFilter.mInputFilter, mInputFilter);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public int hashCode() {
        return Objects.hashCode(mMaxQualifiers, mOffset, mInputFilter);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public String toString() {
        return Objects.toStringHelper(PaginationFilter.class)
            .add("max-qualifiers", mMaxQualifiers)
            .add("offset", mOffset)
            .add("filter", mInputFilter)
            .toString();
    }
}
